package com.shujia.akka.register

import akka.actor.{Actor, ActorRef, ActorSystem, Props}
import com.shujia.akka.clientAndServer.Server.{serverHost, serverPort}
import com.typesafe.config.{Config, ConfigFactory}

import java.util.Date
import java.util.concurrent.TimeUnit
import scala.collection.mutable
import scala.concurrent.duration.FiniteDuration

object Master extends App {
  /**
   * 1.Master和Worker启动之后，需要打印提示信息
   * 2.Worker需要向Master发送注册信息，并提交当前节点的计算资源信息
   * 3.Worker定期发送心跳信息给Master，Master接收信息后会更新心跳信息
   * 4.Master需要定期检查当前Worker是否宕机
   */

  var (masterHost, masterPort) = ("127.0.0.1", 8888)

  private val config: Config = ConfigFactory.parseString(
    s"""
       |akka.actor.provider="akka.remote.RemoteActorRefProvider"
       |akka.remote.netty.tcp.hostname=$masterHost
       |akka.remote.netty.tcp.port=$masterPort
       |""".stripMargin
  )

  private val masterSystem: ActorSystem = ActorSystem("MasterSystem", config)

  private val masterRef: ActorRef = masterSystem.actorOf(Props[Master], "master")

  masterRef ! "start"


}


class Master extends Actor {

  private val hashMap = new mutable.HashMap[String, WorkerInfo]()

  override def receive: Receive = {
    case "start" => {
      println("主节点Master已经初始化完成...")
      self ! "check"
    }
    case WorkerInfo(id, cpu, memory, registerDate, heartbeatDate) => {
      hashMap.put(id, WorkerInfo(id, cpu, memory, registerDate, heartbeatDate))
      println(s"Worker: ${id}已经注册成功...")
    }

    case WorkerHeartbeat(id, heartbeatDate) => {
      println(s"Master接受到${id} 的心跳")
      //      val info: WorkerInfo =
      //      hashMap.put(id, WorkerInfo(info.id, info.cpu, info.memory, info.registerDate,heartbeatDate))
      hashMap(id).heartbeatDate = heartbeatDate
      println(s"${id} 的心跳时间${heartbeatDate.getTime}更新完成...")
    }

    case "check" => {
      println("Master检查机制开始启动...")
      context.system.scheduler.schedule(
        new FiniteDuration(10, TimeUnit.SECONDS)
        , new FiniteDuration(10, TimeUnit.SECONDS)
        , self
        , "check worker heartbeat")(context.dispatcher)
      println("当前定时任务已经启动，每隔10秒检查一次心跳 ")
    }
    case "check worker heartbeat" => {
      println("开始检查所有Worker的心跳")

      for (elem <- hashMap) {
        val currentTime: Long = new Date().getTime
        // 如果间隔时间大于 10秒 对应worker需要去除
        val delayTime: Long = currentTime - elem._2.heartbeatDate.getTime
        println("delayTime:" + delayTime)
        val deathFlag: Boolean = delayTime >= 10 * 1000
        if (deathFlag) {
          hashMap.remove(elem._1)
          println(s"当前${elem._2.id} 已经死亡,在当前存活列表中进行去除...")
        }
      }
    }


  }
}